package wangwenjun.phase2.concurrencypattern.pipeline;

import java.util.stream.IntStream;

/**
 * @author ChangLiang
 * @date 2020/6/17
 */
public class ProductPipeline {

    /**
     * 产品传送带，在传送带上除了负责产品加工的工人之外，还有在传送带上等待加工的产品
     */

    /**
     * 传送带上最多可以有多少个待加工的产品
     */
    private final static int MAX_PRODUCT_COUNT = 100;

    /**
     * 存放代加工的产品
     */
    private final Product[] productQueue;

    private int head;

    private int tail;

    /**
     * 当前在流水线上有多少个待加工的产品
     */
    private int totalProductCount;

    /**
     * 传送带中聚合了worker,传送带知道worker的存在
     */
    private final WorkerThread[] workers;

    public ProductPipeline(int workerSize) {
        this.workers = new WorkerThread[workerSize];
        this.productQueue = new Product[MAX_PRODUCT_COUNT];
        // 实例化每一个工人（ Worker 线程） 并且启动
        IntStream.range(0,workerSize).forEach(i->new WorkerThread("product-pipeline-worker-"+i,this).start());
    }

    /**
     * 接受来自上游的半成品（待加工的产品）
     * @param product
     */
    public void offer(Product product) {
        synchronized (this) {
            while (totalProductCount > MAX_PRODUCT_COUNT) {
                try {
                    this.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            productQueue[tail] = product;
            tail = (tail + 1) % productQueue.length;
            totalProductCount++;
            this.notifyAll();
        }
    }

    public Product take() {
        Product product;
        synchronized (this) {
            while (totalProductCount <= 0) {
                try {
                    this.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            product = productQueue[head];
            head = (head + 1) % productQueue.length;
            totalProductCount--;
            this.notifyAll();
            return product;
        }
    }
}
